package com.gitee.huxing.courese.core;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Description:
 * @ProjectName: JavaCourse
 * @Package: com.gitee.huxing.courese.core
 * @ClassName: CustomMessage
 * @Author: huxing
 * @DateTime: 2021-11-07 5:36 下午
 */
public class CustomMessageQueue<E> extends AbstactMessageQueue<E>{

    /** 最大消息队列数 **/
    private final int capacity;

    /** 原子计数器 **/
    private final AtomicInteger count = new AtomicInteger();

    /** 原子锁 **/
    private final ReentrantLock takeLock = new ReentrantLock();

    /** 读取判断 **/
    private final Condition notEmpty = takeLock.newCondition();

    /** 写入锁 **/
    private final ReentrantLock putLock = new ReentrantLock();

    /** 写入判断 **/
    private final Condition notFull = putLock.newCondition();

    /** 消费者队列 **/
    private ConcurrentHashMap<Integer, Object> consumeQueue = new ConcurrentHashMap<>();

    /** 消息数组 **/
    private ConcurrentHashMap<String, Object> message;


    public CustomMessageQueue(int capacity) {
        this.capacity = capacity;
        this.message = new ConcurrentHashMap<>(capacity);
    }


    @Override
    public boolean add(E e) {
        return false;
    }

    @Override
    public boolean offer(E e) {
        if (e == null){
            throw new NullPointerException();
        }
        final AtomicInteger count = this.count;
        if (count.get() == capacity){
            return false;
        }
        int c = -1;
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            if (count.get() < capacity) {
                this.message.put(count.toString(), e);
                c = count.getAndIncrement();
                if (c + 1 < capacity){
                    notFull.signal();
                }
            }
        } finally {
            putLock.unlock();
        }
        if (c == 0){
            signalNotEmpty();
        }
        return c >= 0;
    }

    @Override
    public E remove() {
        return null;
    }

    @Override
    public E poll(Integer consumerId){
        final AtomicInteger count = this.count;
        if (count.get() == 0){
            return null;
        }
        E x = null;
        int c = -1;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            if (count.get() > 0) {
                x = (E)this.message.get(count.toString());
                // TODO: 在消费的时候写入消费者的编号和消费偏移量
                this.consumeQueue.put(consumerId, count.intValue());
                c = count.getAndDecrement();
                if (c > 1){
                    notEmpty.signal();
                }
            }
        } finally {
            takeLock.unlock();
        }
        if (c == capacity){
            signalNotFull();
        }
        return x;
    }

    public E poll(Integer consumerId, long timeout, TimeUnit unit) throws InterruptedException {
        E x = null;
        int c = -1;
        long nanos = unit.toNanos(timeout);
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                if (nanos <= 0){
                    return null;
                }
                nanos = notEmpty.awaitNanos(nanos);
            }
            x = (E)this.message.get(count.toString());
            // TODO: 在消费的时候写入消费者的编号和消费偏移量
            this.consumeQueue.put(consumerId, count.intValue());
            c = count.getAndDecrement();
            if (c > 1){
                notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }
        if (c == capacity){
            signalNotFull();
        }
        return x;
    }

    private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }

    private void signalNotFull() {
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            notFull.signal();
        } finally {
            putLock.unlock();
        }
    }
}
